package com.yss.netty;

import com.yss.exception.RouterParallelException;
import com.yss.netty.codec.NettyDecode;
import com.yss.netty.codec.NettyEncode;
import com.yss.netty.config.NettyServerConfig;
import com.yss.netty.handler.NettyServerHandler;
import com.yss.netty.processor.AbstractHandlerProcessor;
import com.yss.netty.protocol.CommandType;
import com.yss.netty.util.Constants;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty Server
 */
@Slf4j
public class NettyRemotingServer {

    /**
     * Boss EventLoopGroup
     */
    private final EventLoopGroup bossGroup;

    /**
     * Worker EventLoopGroup
     */
    private final EventLoopGroup workerGroup;

    /**
     * Server BootStrap
     */
    private final ServerBootstrap serverBootstrap;

    /**
     * Start Flag
     */
    private final AtomicBoolean startFlag;

    /**
     * Server Config
     */
    private final NettyServerConfig serverConfig;

    /**
     * Processors' default thread pool
     */
    private final ExecutorService defaultExecutorService;

    /**
     * Handler in pipeline
     */
    private final NettyServerHandler serverHandler;

    public NettyRemotingServer(final NettyServerConfig serverConfig) {
        this.serverConfig = serverConfig;
        this.serverBootstrap = new ServerBootstrap();
        this.startFlag = new AtomicBoolean(false);
        this.serverHandler = new NettyServerHandler();
        this.defaultExecutorService = Executors.newFixedThreadPool(Constants.CPUS);

        bossGroup = new NioEventLoopGroup(this.serverConfig.getBossThreadNum(), new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyServerBossThread_%d", this.threadIndex.incrementAndGet()));
            }
        });

        workerGroup = new NioEventLoopGroup(this.serverConfig.getWorkerThreadNum(), new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyServerWorkerThread_%d", this.threadIndex.incrementAndGet()));
            }
        });

    }

    public void start() {
        if (startFlag.compareAndSet(false, true)) {
            this.serverBootstrap
                    .group(this.bossGroup, this.workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_REUSEADDR, this.serverConfig.SO_REUSEADDR)
                    .option(ChannelOption.SO_BACKLOG, this.serverConfig.getSoBacklog())
                    .option(ChannelOption.SO_KEEPALIVE, this.serverConfig.SO_KEEPALIVE)
                    .option(ChannelOption.TCP_NODELAY, this.serverConfig.isTcpNoDelay())
                    .option(ChannelOption.SO_SNDBUF, this.serverConfig.getSendBufferSize())
                    .option(ChannelOption.SO_RCVBUF, this.serverConfig.getReceiveBufferSize())
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            initPipelineHandle(ch);
                        }
                    });
            ChannelFuture future;
            try {
                future = serverBootstrap.bind(this.serverConfig.getListenerPort()).sync();
            } catch (Exception e) {
                log.error("\nNettyRemotingServer bind failed：{}", e.getMessage());
                throw new RouterParallelException(e);
            }
            if (future.isSuccess()) {
                log.info("\nNettyRemotingServer bind {} success", this.serverConfig.getListenerPort());
            } else {
                throw new RouterParallelException(String.format("\nNettyRemotingServer bind failed：%s", future.cause().toString()));
            }
        }
    }

    /**
     * Register with default thread pool
     *
     * @param commandType
     * @param processor
     */
    public void registerProcessors(CommandType commandType, AbstractHandlerProcessor processor) {
        serverHandler.registerProcessor(commandType, processor, defaultExecutorService);
    }

    /**
     * Register with given thread pool
     *
     * @param commandType
     * @param processor
     * @param executorService
     */
    public void registerProcessors(CommandType commandType, AbstractHandlerProcessor processor, ExecutorService executorService) {
        serverHandler.registerProcessor(commandType, processor, executorService);
    }


    private void initPipelineHandle(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("decoder", new NettyDecode())
                .addLast("encoder", new NettyEncode())
                .addLast("server-idle-handler", new IdleStateHandler(0, 0, Constants.NETTY_SERVER_HEART_BEAT_TIME, TimeUnit.MILLISECONDS))
                .addLast("netty-server-handler", serverHandler);
    }

    public void close() {
        if (this.startFlag.compareAndSet(true, false)) {
            if (this.bossGroup != null) {
                this.bossGroup.shutdownGracefully();
            }
            if (this.workerGroup != null) {
                this.workerGroup.shutdownGracefully();
            }
            log.info("\nNetty server closed");
        }
    }


}
